package com.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author : ranzlupup
 * @date : 2023/3/1 16:14
 */
object RDD_Transform_KV_aggregateByKey {
    def main(args: Array[String]): Unit = {
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
        val sparkContext: SparkContext = new SparkContext(sparkConf)
        val dataRDD1: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3), ("a", 2), ("c", 5), ("a", 4)), 2)

        // TODO aggregateByKey
        //      将数据根据不同的规则进行分区内计算和分区间计算
        //      函数柯里化，存在两个参数列表
        //      第一个参数：
        //          列表中的参数表示初始值
        //      第二个参数：
        //          第一个函数，写分区内的规则
        //          第二个函数，写分区间的规则
        dataRDD1.aggregateByKey(0)(
            (x, y) => math.max(x, y),
            (x, y) => x + y
        )
            .collect()
            .foreach(println)

        println("=============================")
        sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3), ("a", 2), ("c", 5), ("a", 4)), 2).aggregateByKey((0, 0))(
            (t, v) => {
                (t._1 + v, t._2 + 1)
            },
            (t1, t2) => {
                (t1._1 + t2._1, t1._2 + t2._2)
            }
        )
            .mapValues{
                case (num, time) => num / time
            }
            .collect().foreach(println)
        sparkContext.stop()
    }
}
